package com.ubxtech.rocketmq.aspect;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.StopWatch;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @author Victor.Xiao
 * @since 2022-04-27 16:28
 **/

@Component
@Slf4j
@Aspect
public class RocketMqConsumerAspect {

    @Pointcut("execution(void org.apache.rocketmq.spring.core.RocketMQListener.onMessage(..))")
    public void cutMethod(){}

    @Around("cutMethod()")
    public Object invoke(ProceedingJoinPoint pjp){

        StopWatch clock = new StopWatch();
        Object[] args = pjp.getArgs();
        clock.start();
        MethodSignature ms = (MethodSignature) pjp.getSignature();
        RocketMQMessageListener annotation = ms.getMethod().getDeclaringClass().getAnnotation(RocketMQMessageListener.class);
        String consumerGroup = annotation.consumerGroup();
        String uuid = UUID.randomUUID().toString();
        try {
            if (log.isDebugEnabled()) {
                log.debug("收到消息:uuid:{},topic:{}, group:{},msg:{}", uuid,annotation.topic(),consumerGroup,args[0]);
            }
            pjp.proceed(args);
        } catch (Throwable throwable) {
            log.error("消息消费异常=============>topic:{}, group:{},uuid:{},msg:{}", annotation.topic(),consumerGroup,uuid,args[0]);
            log.error(throwable.getMessage(), throwable);
        } finally {
            clock.stop();
            if (log.isDebugEnabled()) {
                log.debug("消息消费结束  uuid: {}, 耗时:{}", uuid,clock.getTime());
            }
        }
        return null;
    }
}
